/* =========================================================================
 *
 *   The traffic processor.
 *
 *   Development by Sedelnikov Alexander (Bigking).
 *
 * =========================================================================
 */
// @todo Make pre strings and poststring to be outputted!

#include "job_manager.h"
#include "log.h"
#include <stdlib.h>

#include <sys/stat.h>
#include <dirent.h>
#include <string.h>
#include <math.h>

#include "ipacc3_parser.h"
#include "out.h"
#include "transact.h"


//-----------------------------------------------------------------------//
cjob_manager   cjob_manager::singleton = cjob_manager();


//-----------------------------------------------------------------------//
cjob_manager::cjob_manager()
{
}


//-----------------------------------------------------------------------//
cjob_manager::~cjob_manager()
{
}


//-----------------------------------------------------------------------//
int check_legal_source_file( const char *file_name )
{
    FILE       *fp;

    char        full_name[ FILENAME_MAX * 2 ];
    const char *file_mask = cconfig::get_files_mask();

    int         l_str  = strlen( file_name ),
                l_mask = strlen( file_mask );

    if( l_mask >= l_str )
        return 0;

    sprintf( full_name, "%s/%s.downloaded",
             cconfig::get_source_folder(), file_name );

    if( ( fp = fopen( full_name, "rb" ) ) == NULL )
        return 0;

    fclose( fp );

    for( int i = l_str - 1, j = l_mask - 1;
         j >= 0; i--, j-- )
        if( file_name[ i ] != file_mask[ j ] )
            return 0;

    return 1;
}


//-----------------------------------------------------------------------//
cparser* cjob_manager::get_parser()
{
    cparser      *parser;

    switch( cconfig::get_files_type() & FILE_TYPE_MASK )
    {
        case FILE_TYPE_IPACC3:
            parser = new cipacc3_parser();
        break;
        default:
            THROW( e_error, "Unknown type files to be parsed!" );
            return NULL;
    }

    return  parser;
}

//-----------------------------------------------------------------------//
cfile_base* cjob_manager::get_input()
{
    cfile_base   *parser_input;

    switch( cconfig::get_files_type() >> 16 )
    {
        case ( FILE_TYPE_GZ >> 16 ):
            parser_input = new cgzfile();
        break;
        default:
            parser_input = new cfile();
        break;
    }

    return  parser_input;
}


//-----------------------------------------------------------------------//
int cjob_manager::do_job( )
{
    logmsg( "Start job..." );

    singleton.scan_source_folder();

    if( singleton.files.size() == 0 )
    {
        logmsg( "Nothing to do. Empty source folder!" );
        return 0;
    }

    singleton.interval_distribute();
    singleton.prepare_processors();
    singleton.process_source();
    singleton.save_result();

    return 0; // @todo remove it!
}


//-----------------------------------------------------------------------//
void cjob_manager::scan_source_folder()
{
    logdbg( "Scan source folder..." );
    //
    // Get files from source.
    //
    char         full_path[ 2048 ];

    sprintf( full_path, "%s/", cconfig::get_source_folder() );

    DIR         *dir_ptr  = opendir( full_path );
    dirent      *dir_item = NULL;

    if( dir_ptr == NULL )
    {
        logerr( "Source directory [%s] not exist!", full_path );
        return;
    }

    cparser     *parser_time       = get_parser();
    cfile_base  *parser_input_time = get_input();

    files.clear();

    while( ( dir_item = readdir( dir_ptr ) ) != NULL )
    {
        if( check_legal_source_file( dir_item->d_name ) )
        {
            source_file     sf;
            strcpy( sf.file_name, dir_item->d_name );

            sprintf( full_path, "%s/%s",
                     cconfig::get_source_folder(),
                     sf.file_name );

            parser_time->set_input( parser_input_time );
            parser_time->start( full_path );
            sf.stamp = parser_time->get_time( full_path );
            parser_time->end( );
            sf.interval = sf.stamp.to_minute();
            files.push_back( sf );
        }
    }

    delete parser_time;
    delete parser_input_time;

    closedir( dir_ptr );

    int start_interval = 0;

    if( files.size() )
    {
        int     index    = 0,
                interval = 0;

        time_interval    = 0;

        sort( files.begin(), files.end(), source_file::sort_cmp_func );

        start_interval   = files[ 0 ].interval;

        for( index = 0; index < files.size(); index++ )
        {
            if( files[ index ].interval - start_interval <
                cconfig::get_time_interval_min() )
                continue;

            if( files[ index ].interval - start_interval >
                cconfig::get_time_interval_max() )
                break;
        }

        interval = files[ index - 1 ].interval - start_interval;
        if( interval > cconfig::get_time_interval_max() )
            interval = cconfig::get_time_interval_max();

        float  th_min = (float)cconfig::get_threads_count_min(),
               th_mid = ( (float)cconfig::get_threads_count_max() -
                          (float)cconfig::get_threads_count_min() )
                          / 2.0f * cconfig::get_distrib_low(),
               th_max = (float)cconfig::get_threads_count_max(),
               s = (float) interval / (float) cconfig::get_time_interval_max();


        th_cnt = (float)( th_min * ( 1.0 - s ) * ( 1.0 - s ) +
                          th_mid * 2.0 * s * ( 1.0 - s )     +
                          th_max * s * s ) * 1.3f;

        if( th_cnt < cconfig::get_threads_count_min() )
            th_cnt = cconfig::get_threads_count_min();
        if( th_cnt > cconfig::get_threads_count_max() )
            th_cnt = cconfig::get_threads_count_max();

        time_interval = ( interval + 4 ) / th_cnt;

        if( time_interval < cconfig::get_time_interval_min() )
            time_interval = cconfig::get_time_interval_min();

        // @fixme Is temporary solution.
        time_interval = ( time_interval + 4 ) / 5;
        time_interval *= 5;
        if( time_interval == 0 )
            time_interval = 5;

//        if( ( 60 % time_interval ) != 0 )
//        {
           if( time_interval == 25 )
               time_interval = 20;

           if( time_interval == 35 )
               time_interval = 30;

           if( time_interval == 40 || time_interval == 45 )
               time_interval = 30;

           if( time_interval == 50 || time_interval == 55 )
               time_interval = 60;

           if( time_interval > 60 )
               time_interval = 60;

           if( time_interval == 0 )
               time_interval = 5;
//        }

        logdbg( "Dynamic interval = %ld", time_interval );

        for( index = 0; index < files.size(); index++ )
        {
            files[ index ].stamp_trunc.from_minute(
                ( files[ index ].interval / time_interval ) *
                time_interval );

            files[ index ].interval_trunc = ( files[ index ].interval / time_interval ) *
                time_interval;
        }

    }

    logdbg( "[%ld] files was scaned.", files.size() );
}


//-----------------------------------------------------------------------//
void cjob_manager::interval_distribute( )
{
    //
    // At this time, processor manager distribute
    // intervals for files per threads
    //
    logdbg( "Interval distribution..." );

    memset( procs_indexes, 0, sizeof( ulong ) * MAX_THREADS );
    memset( procs_lengths, 0, sizeof( ulong ) * MAX_THREADS );

    ulong         start_interval         = files[ 0 ].interval_trunc,
                  start_index            = 0,
                  cur_interval           = files[ 0 ].interval,
                  cur_index              = 0;

    for( int i = 0; i < th_cnt; i++ )
    {
        start_interval = files[ start_index ].interval_trunc,
        cur_index      = start_index;
        cur_interval   = files[ cur_index ].interval;

        while( cur_interval - start_interval < time_interval )
        {
            if( cur_index >= files.size() )
                break;

            if( files[ cur_index ].interval - start_interval >= time_interval )
                break;

            cur_interval = files[ cur_index++ ].interval;
        }

        procs_indexes[ i ] = start_index;
        procs_lengths[ i ] = cur_index - start_index;

        logdbg( "index=%ld length=%ld st_i=%ldminute cur_i=%ldminute",
                procs_indexes[ i ], procs_lengths[ i ],
                start_interval, cur_interval );

        if( cur_index >= files.size() )
            break;

        start_index = cur_index;
    }

    //
    // Debug
    //
    logdbg( "Distribution:" );
    for( int i = 0; i < th_cnt; i++ )
    {
        for( int j = 0; j < procs_lengths[ i ]; j++ )
        {
            logdbg( "th=[%ld] fn=[%s] ts=[%04ld-%02ld-%02ld %02ld:%02ld:00]/[%04ld-%02ld-%02ld %02ld:%02ld:00]",
                     i, files[ procs_indexes[ i ] + j ].file_name,
                     files[ procs_indexes[ i ] + j ].stamp.year,
                     files[ procs_indexes[ i ] + j ].stamp.month,
                     files[ procs_indexes[ i ] + j ].stamp.day,
                     files[ procs_indexes[ i ] + j ].stamp.hour,
                     files[ procs_indexes[ i ] + j ].stamp.minute,
                     files[ procs_indexes[ i ] + j ].stamp_trunc.year,
                     files[ procs_indexes[ i ] + j ].stamp_trunc.month,
                     files[ procs_indexes[ i ] + j ].stamp_trunc.day,
                     files[ procs_indexes[ i ] + j ].stamp_trunc.hour,
                     files[ procs_indexes[ i ] + j ].stamp_trunc.minute
                     );
        }
    }

    logdbg( "Distribution complite." );
}


//-----------------------------------------------------------------------//
void cjob_manager::prepare_processors( )
{
    //
    // Prepare parser / Hash tables
    //
    logdbg( "Prepare processor(s)..." );

    parsers       = new cparser*[ th_cnt ];
    inputs        = new cfile_base*[ th_cnt ];
    hashes        = new chash_table<trf_gth>*[ th_cnt ];

    for( int i = 0; i < th_cnt; i++ )
    {
        hashes[ i ] = new chash_table<trf_gth>();

        if( hashes[ i ]->create( 1 << cconfig::get_hash_size(),
                                 64 / cconfig::get_hash_size() + 1 ) != 0 )
            THROW( e_error, "Can't allocate a memory for hash table." );


        parsers[ i ]   = get_parser();
        inputs[ i ]    = get_input();
        parsers[ i ]->set_input( inputs[ i ] );
    }

    logdbg( "Processors are prepared." );
}


//-----------------------------------------------------------------------//
void cjob_manager::process_source( )
{
    //
    // Threads prepare
    //
    logdbg( "Source processing..." );

    char         full_path[ 2048 ];

    cprocessor            *procs = new cprocessor[ th_cnt ];

    for( int i = 0; i < th_cnt; i++ )
    {
        procs[ i ].set_hash_table( hashes[ i ] );
        procs[ i ].set_parser( parsers[ i ] );
    }

    int                    indexes[ MAX_THREADS ];
    bool                   full_job_complite;

    for( int i = 0; i < th_cnt; i++ )
        indexes[ i ] = procs_indexes[ i ];

    //
    // The main job code
    //
    while( true )
    {
        for( int i = 0; i < th_cnt; i++ )
        {
            if( procs[ i ].complite() )
            {
                if( indexes[ i ] - procs_indexes[ i ] < procs_lengths[ i ] )
                {
                    sprintf( full_path, "%s/%s", cconfig::get_source_folder(),
                             files[ indexes[ i ] ].file_name );

                    try
                    {
                        parsers[ i ]->start( full_path );
                        logdbg( "[%s] - processing...", full_path ); // @todo make more informatable
                    }
                    catch( cexception &exc )
                    {
                        logfatal( "%s", exc.get_formated_message() );
                        return; // @todo make throw
                    }

                    procs[ i ].thread_close();
                    procs[ i ].thread_run();
                    indexes[ i ]++;
                }
            }
        }

        //
        // Check a job state. May be it complite?
        //
        full_job_complite = true;

        for( int i = 0; i < th_cnt; i++ )
            if( !procs[ i ].complite()
                 || indexes[ i ] - procs_indexes[ i ] < procs_lengths[ i ] )
            {
                full_job_complite = false;
                break;
            }

        //
        // So... job is complite, then exit
        //
        if( full_job_complite )
        {
            break;
        }

#ifdef  _WIN32
        Sleep( 10 );
#else
        usleep( 100 );
#endif
    }

    //
    // Close all threads.
    //
    // @todo See for linux thread close function (pthread_close ?)
    //
    for( int i = 0; i < th_cnt; i++ )
    {
        procs[ i ].thread_close();

        parsers[ i ]->end();
        delete parsers[ i ];
        delete inputs [ i ];
    }

    delete []parsers;
    delete []procs;
    delete []inputs;

    logdbg( "Source is processed." );
}


//-----------------------------------------------------------------------//
void cjob_manager::save_result( )
{
    //
    // Now output results from the thread(s).
    //
    logdbg( "Save results..." );

    char                   text[ 2048  ],
                       path_arh[ 2048 ],
                       path_src[ 2048 ],
                       path_dst[ 2048 ];

    ctransaction       transaction;

    //
    // Prepare out stream.
    //
    cout_info         *out_info;
    ulong              flt_vi, flt_vo;

    enum
    {
        STREAM_FILE     = 0,
        STREAM_PGSQL
    };

    uint               stream_type = STREAM_FILE;


    if( !strcmp( cconfig::get_out_stream(), "file" ) )
    {
        stream_type = STREAM_FILE;
        out_info    = new cout_file();
    }
    else
    if( !strcmp( cconfig::get_out_stream(), "pgsql" ) )
    {
        stream_type = STREAM_PGSQL;
        out_info    = new cout_postgres();
    }
    else
    {
        THROW( e_error, "Unknown stream type" );
    }


    flt_vi = cconfig::get_out_filter_volume_in();
    flt_vo = cconfig::get_out_filter_volume_out();


    for( int th_index = 0;
             th_index < th_cnt;
             th_index++ )
    {
        if( hashes[ th_index ]->size() != 0 )
        {

            //
            // First!
            // Make arhive copy
            //
            for( int i = 0;
                     i < procs_lengths[ th_index ];
                     i++ )
            {

                if( !strcmp( cconfig::get_arhive_collect(), "day" ) )
                {
                    sprintf( path_arh, "%s/%04ld_%02ld_%02ld",
                             cconfig::get_arhive_folder(),
                             files[ procs_indexes[ th_index ] + i ].stamp.year,
                             files[ procs_indexes[ th_index ] + i ].stamp.month,
                             files[ procs_indexes[ th_index ] + i ].stamp.day );
                }
                else
                {
                    sprintf( path_arh, "%s/%04ld_%02ld",
                             cconfig::get_arhive_folder(),
                             files[ procs_indexes[ th_index ] + i ].stamp.year,
                             files[ procs_indexes[ th_index ] + i ].stamp.month );
                }
#ifndef _WIN32
                mkdir( cconfig::get_arhive_folder(), 0777 );
                mkdir( path_arh, 0777 );
#else
                mkdir( cconfig::get_arhive_folder() );
                mkdir( path_arh );
#endif

                sprintf( path_src, "%s/%s",
                         cconfig::get_source_folder(),
                         files[ procs_indexes[ th_index ] + i ].file_name );

                sprintf( path_dst, "%s/%s",
                         path_arh,
                         files[ procs_indexes[ th_index ] + i ].file_name );

                if( file_exist( path_dst ) )
                {
                    if( file_size( path_src ) != file_size( path_dst ) )
                    {
                        logwrn( "Arhive filename collision [%s]->[%s]", path_src, path_dst );

                        while( file_exist( path_dst ) )
                        {
                            sprintf( path_dst, "%s/collide_%04X%04X_%s",
                                     path_arh,
                                     rand(),
                                     rand(),
                                     files[ procs_indexes[ th_index ] + i ].file_name
                                   );
                        }
                    }
                }

                file_copy( path_src, path_dst );

                logdbg( "Copy to arhive [%s] -> [%s]", files[ procs_indexes[ th_index ] + i ].file_name, path_dst );

                transaction.append_file( path_src, path_dst );
            }

            //
            // Start saving result with transaction...
            //
            transaction.begin();

            chash_table<trf_gth>::iterator           iter;

            sprintf( text, "%04ld_%02ld_%02ld_%02ld_%02ld",
                     files[ procs_indexes[ th_index ] ].stamp.year,
                     files[ procs_indexes[ th_index ] ].stamp.month,
                     files[ procs_indexes[ th_index ] ].stamp.day,
                     files[ procs_indexes[ th_index ] ].stamp.hour,
                     files[ procs_indexes[ th_index ] ].stamp.minute );

            logdbg( "Save from hash table..." );

            out_info->set_label( text );
            out_info->start();


            trf_gth trf;

            for( std::vector<std::string>::const_iterator iter =
                cconfig::get_pre_cmd_begin();
                iter != cconfig::get_pre_cmd_end();
                iter++ )
                out_info->exec( iter->c_str() );

            int  v = 0;

            for( iter = *hashes[ th_index ];
                 iter;
                 iter++ )
            {
                trf = iter;
                v++;

                // @todo prepare
                if( trf.volume_in  < flt_vi &&
                    trf.volume_out < flt_vo )
                    continue;

                cout_info::format_traffic( trf,
                                 files[ procs_indexes[ th_index ] ].stamp_trunc,
                                 cconfig::get_out_format_line(),
                                 text,
                                 1000 );

                out_info->send_line( text, 1000 );
            }

            logdbg( "Sended [%ld] in hash [%ld].", v, hashes[ th_index ]->size() );

            for( std::vector<std::string>::const_iterator iter =
                cconfig::get_post_cmd_begin();
                iter != cconfig::get_post_cmd_end();
                iter++ )
                out_info->exec( iter->c_str() );


            //
            //
            //
            logdbg( "Remove files from the source folder." );

            for( int i = 0;
                     i < procs_lengths[ th_index ];
                     i++ )
            {
                sprintf( path_src, "%s/%s",
                         cconfig::get_source_folder(),
                         files[ procs_indexes[ th_index ] + i ].file_name );

                if( cconfig::debug_traffic_src_not_remove() == false )
                {
                    remove( path_src );
                    sprintf( path_dst, "%s/%s.downloaded",
                             cconfig::get_source_folder(),
                             files[ procs_indexes[ th_index ] + i ].file_name );
                    remove( path_dst );
                }
            }

            out_info->end( &transaction );

            transaction.end();
            out_info->finish();

            hashes[ th_index ]->release();
            delete hashes[ th_index ];

            logdbg( "Results of thread/processor [%ld] is saved.", th_index );
        }
        else
        {
            logdbg( "Thread/processor [%ld] is empty!", th_index );
        }

    }

    delete out_info;
    delete []hashes;

    logdbg( "Results are saved." );
}
